package com.ikas.ai.server.module.alarm.service;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.ikas.ai.model.KafkaMeteData;
import com.ikas.ai.server.error.BusinessException;
import com.ikas.ai.server.kafka.RedisDataOperation;
import com.ikas.ai.server.kafka.WorkInstanceService;
import com.ikas.ai.server.module.alarm.dao.TripAlarmDAO;
import com.ikas.ai.server.module.alarm.model.TripAlarm;
import com.ikas.ai.server.module.alarm.model.enums.AlarmType;
import com.ikas.ai.server.module.condition.enums.WorkFlowTypeEnum;
import com.ikas.ai.server.module.data.model.TripMete;
import com.ikas.ai.server.module.data.service.TripMeteService;
import com.ikas.ai.server.module.instance.enums.WorkFlowInstanceStateEnum;
import com.ikas.ai.server.module.instance.enums.WorkInstanceStepDetailStateEnum;
import com.ikas.ai.server.module.instance.enums.WorkInstanceStepStateEnum;
import com.ikas.ai.server.module.instance.model.DorisWorkFlowInstance;
import com.ikas.ai.server.module.instance.model.DorisWorkInstanceStep;
import com.ikas.ai.server.module.instance.model.DorisWorkInstanceStepDetail;
import com.ikas.ai.server.module.instance.service.DorisWorkFlowInstanceService;
import com.ikas.ai.server.module.instance.service.DorisWorkInstanceStepDetailService;
import com.ikas.ai.server.module.instance.service.DorisWorkInstanceStepService;
import com.ikas.ai.utils.CollectionUtil;
import com.ikas.ai.utils.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
@Slf4j
public class TripAlarmService extends ServiceImpl<TripAlarmDAO, TripAlarm> {

    @Autowired
    private RedisDataOperation initDataRedisListener;

    @Autowired
    private DorisWorkFlowInstanceService dorisWorkFlowInstanceService;
    @Autowired
    private DorisWorkInstanceStepDetailService dorisWorkInstanceStepDetailService;
    @Autowired
    private DorisWorkInstanceStepService dorisWorkInstanceStepService;
    @Autowired
    private WorkInstanceService workInstanceService;
    @Autowired
    AlarmCacheService alarmCacheService;
    @Autowired
    TripMeteService tripMeteService;

    /**
     * 当工况流程在执行中收到跳机信号后，全部应该都变为已执行失败
     *
     * @param mete
     */
    @Transactional(rollbackFor = Exception.class)
    public void processTripMete(KafkaMeteData mete) {
        TripMete tripMete = initDataRedisListener.getCacheTripMete(mete.getMachineNo(), mete.getMeteCode());
        if (null == tripMete) {
            log.error("从缓存中未获取到跳机测点");
            return;
        }
        List<DorisWorkFlowInstance> instanceList = dorisWorkFlowInstanceService.getProcessingList(mete.getMachineNo());
        if (CollectionUtil.isEmpty(instanceList)) {
            log.info("收到跳机测点,未有运行中的工况");
        }
        insertAlarm(mete, instanceList);
        updateWorkFlowInstanceAndTheirStepsAndDetails(mete, tripMete, instanceList);
    }


    private void updateWorkFlowInstanceAndTheirStepsAndDetails(KafkaMeteData mete, TripMete tripMete, List<DorisWorkFlowInstance> instanceList) {
        instanceList = instanceList.stream().filter(p->!WorkFlowTypeEnum.TB70.getName().equalsIgnoreCase(p.getWfName())).collect(Collectors.toList());
        if (CollectionUtil.isEmpty(instanceList)) {
            return;
        }
        List<Long> instanceIds = instanceList.stream().map(DorisWorkFlowInstance::getId).collect(Collectors.toList());
        Map<Long, DorisWorkFlowInstance> instanceMapById = instanceList.stream().collect(Collectors.toMap(DorisWorkFlowInstance::getId, bean -> bean));
        List<DorisWorkInstanceStep> stepList = dorisWorkInstanceStepService.list(
                new LambdaQueryWrapper<DorisWorkInstanceStep>()
                        .in(CollectionUtils.isNotEmpty(instanceIds), DorisWorkInstanceStep::getWfInstanceId, instanceIds)
                        .in(DorisWorkInstanceStep::getState, WorkInstanceStepStateEnum.NOT_STARTED.getCode(), WorkInstanceStepStateEnum.PROCESSING.getCode())
                        .orderByAsc(DorisWorkInstanceStep::getSort)
        );
        List<Long> stepIds = stepList.stream().map(DorisWorkInstanceStep::getId).collect(Collectors.toList());
        List<DorisWorkInstanceStepDetail> stepDetailList = dorisWorkInstanceStepDetailService.list(
                new LambdaQueryWrapper<DorisWorkInstanceStepDetail>().in(DorisWorkInstanceStepDetail::getWfInstanceId, instanceIds)
                        .in(CollectionUtils.isNotEmpty(stepIds), DorisWorkInstanceStepDetail::getStepInstanceId, stepIds)
                        .eq(DorisWorkInstanceStepDetail::getState, WorkInstanceStepDetailStateEnum.NOT_STARTED.getCode())
        );

        List<DorisWorkFlowInstance> instancesUpdateList = instanceList.stream().map(instance -> {
            instance.setEndTime(mete.getGenerateTime());
            instance.setCostTime(Math.abs(instance.getEndTime() - instance.getStartTime()));
            instance.setState(WorkFlowInstanceStateEnum.ERROR.getCode());
            instance.setDesc("收到跳机信号，异常结束," + JSONUtil.toJsonStr(mete) + "," + JSONUtil.toJsonStr(tripMete));
            instance.setUpdateTime(new Date());
            return instance;
        }).collect(Collectors.toList());

        List<DorisWorkInstanceStep> stepsUpdateList = stepList.stream().map(step -> {
            step.setState(WorkInstanceStepStateEnum.ERROR.getCode());
            step.setUpdateTime(new Date());
            workInstanceService.saveOrUpdateMachineInfo(instanceMapById.get(step.getWfInstanceId()), step);
            return step;
        }).collect(Collectors.toList());

        List<DorisWorkInstanceStepDetail> detailUpdateList = stepDetailList.stream().map(detail -> {
            detail.setState(WorkInstanceStepDetailStateEnum.ERROR.getCode());
            detail.setUpdateTime(new Date());
            return detail;
        }).collect(Collectors.toList());

        dorisWorkFlowInstanceService.saveBatch(instancesUpdateList);
        dorisWorkInstanceStepService.saveBatch(stepsUpdateList);
        dorisWorkInstanceStepDetailService.saveBatch(detailUpdateList);
    }

    private void insertAlarm(KafkaMeteData mete, List<DorisWorkFlowInstance> instanceList) {
        TripMete tripMete = tripMeteService.getOne(new LambdaQueryWrapper<TripMete>().eq(TripMete::getMeteCode, mete.getMeteCode()));
        JSONArray desc = new JSONArray();
        for (DorisWorkFlowInstance instance : instanceList) {
            HashMap<String, String> descInstanceMap = new HashMap<>();
            descInstanceMap.put("wf_name", instance.getWfName());
            descInstanceMap.put("wf_instance_id", String.valueOf(instance.getId()));
            descInstanceMap.put("wc_name", instance.getCurrWc());
            desc.add(descInstanceMap);
        }
        TripAlarm tripAlarm = new TripAlarm()
                .setMachineNo(mete.getMachineNo())
                .setSolved(TripAlarm.ALARM_NOT_SOLVED)
                .setTripTime(mete.getGenerateTime())
                .setTripDate(mete.getGenerateDate())
                .setMeteCode(tripMete.getMeteCode())
                .setMeteName(tripMete.getSignalCnName())
                .setTripMeteType(tripMete.getTripType())
                .setTripMeteId(tripMete.getId())
                .setDescription(desc.toJSONString());
        save(tripAlarm);
        int count = count(
                new LambdaQueryWrapper<TripAlarm>()
                        .eq(TripAlarm::getSolved, TripAlarm.ALARM_NOT_SOLVED)
                        .eq(TripAlarm::getMachineNo, mete.getMachineNo())
        );
        alarmCacheService.updateAlarm(AlarmType.TRIP, tripMete.getMachineNo(), count);
    }
}
